package com.cml.infrastructure;

import com.cml.common.constant.NodeStatus;
import com.cml.common.context.ServerContext;
import com.cml.domain.entity.Message;
import com.cml.domain.entity.ServerNode;
import com.cml.domain.factory.MessageBuilderFactory;
import com.cml.domain.service.ServerNodeHAManagerService;
import org.apache.log4j.Logger;

import java.io.DataInputStream;
import java.io.IOException;
import java.net.Socket;
import java.util.concurrent.BlockingQueue;

/**
 * 姓名：陈茂林
 * 时间：2023/10/7 16:19
 * 描述：
 */
public class ServerReadIoThread extends Thread{
    private Socket socket;
    private BlockingQueue<Message> blockingQueue;

    private DataInputStream dataInputStream;

    private ServerContext serverContext;

    private ServerNode remoteServerNode;

    private IOThreadRunningSignal ioThreadRunningSignal;

    Logger logger = Logger.getLogger(ServerReadIoThread.class);
    public ServerReadIoThread(IOThreadRunningSignal ioThreadRunningSignal,ServerContext serverContext, ServerNode remoteServerNode, Socket socket, BlockingQueue blockingQueue) throws IOException {
        this.socket = socket;
        this.blockingQueue = blockingQueue;
        this.dataInputStream= new DataInputStream(socket.getInputStream());
        this.serverContext = serverContext;
        this.remoteServerNode=remoteServerNode;
        this.ioThreadRunningSignal=ioThreadRunningSignal;
    }

    @Override
    public void run() {
        while(NodeStatus.getInstance().isRunning()&&ioThreadRunningSignal.isRunning()) {
            try {
                Message message = MessageBuilderFactory.buildMessage(this.socket);
                if(message != null){
                    this.blockingQueue.put(message);
                }
            }
            catch (Exception e) {
                logger.error(e);
                logger.error(this.socket.getRemoteSocketAddress() + ":" +this.socket.getPort()+"断开连接！");
                logger.info("开始进行数据清理！");
                ioThreadRunningSignal.setIsRunning(false);
                ServerNodeHAManagerService.getInstance(serverContext).ha(remoteServerNode);
                //做端口的各种逻辑
                //各类资源清理
                //选举
                //槽分配
                //副本生效
                //NodeStatus.getInstance().setCurStatus(NodeStatus.NODE_FAIL);
            }
        }
    }
}
